import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Vector;


public class Client extends Machine implements IClient {

	ICache cache = null;
	IMachine master = null;
	ConnectionManager masterConnection = null,slaveConnection = null;
	Vector<KeySpaceObject> keySpace = null;
	MessageDigest md = null;
	
	public Client() 
	{
		try {
			md = MessageDigest.getInstance("sha-1");
		} catch (NoSuchAlgorithmException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		keySpace = new Vector<KeySpaceObject>();
		cache = new CacheManager();
		master = new Machine();
		master.setIp("141.212.110.104");
		master.setId("MASTER");
		masterConnection = new ConnectionManager(PORTS.MASTER_LISTENER_PORT);
		masterConnection.connectTo(master);
		slaveConnection = new ConnectionManager(PORTS.SLAVE_LISTENER_PORT_FOR_CLIENT);
		//connManager.connectTo(master);		
	}
	
	public void populateKeySpace(byte[] data)
	{
		this.keySpace.removeAllElements();
		ByteArrayInputStream bis = new ByteArrayInputStream(data);
		int total = bis.read();
		for(int i = 0;i < total;i++){
			int point = bis.read();
			int strlen = bis.read();
			byte[] arr = new byte[strlen];
			bis.read(arr, 0, strlen);
			String ip = new String(arr);
			
			int portlen = bis.read();
			byte[] arr2 = new byte[portlen];
			bis.read(arr2, 0, portlen);
			String port = new String(arr2);
			KeySpaceObject ob = new KeySpaceObject(point, ip,port);
			this.keySpace.add(ob);
		}
	}
	
	public KeySpaceObject findSlave(int hashedKey)
	{
		int prev = 0;
		for(int i = 0;i < keySpace.size();i++){
			if(hashedKey > prev && hashedKey < keySpace.get(i).getPoint())
				return keySpace.get(i);
			prev = keySpace.get(i).getPoint();
				
		}
		return keySpace.get(0);
	}
	
	@Override
	public DataObject getData(DataObject key) {
		int hashedKey = getHash(key);		
		String slave = cache.get(key);
		if(slave == null){
			//need to contact the master and get the corresponding slave
			IMessage message = new Message();
			message.setSource(this);
			message.setDest(master);
			message.setMessageID(IMessage.CLIENT_INQUIRE);
			message.setKey(key);		
			message.setValue(new DataObject("".getBytes()));
			
			masterConnection.writeData(master, message.serialize());
			IMessage response = null;
			try {
				response = masterConnection.receiveMessage();
				//populate keyspace
				populateKeySpace(response.getValue().getVal());
			}
			catch (IOException e) {
				e.printStackTrace();
			}
			KeySpaceObject sl = findSlave(hashedKey);
			slave = sl.getSlaveip();
			cache.put(key, sl.getSlaveip());
			((CacheManager)cache).putPort(sl.getSlaveip(), sl.getPort());
		}
		//we should have got the slave by now. now contact the slave directly
		IMessage ms = new Message();
		ms.setSource(this);		
		ms.setMessageID(IMessage.GETDATA);
		ms.setKey(key);
		ms.setValue(new DataObject("".getBytes()));
		IMachine dest = new Machine();
		dest.setIp(slave);
		ms.setDest(dest);
		slaveConnection.connectTo(dest,new Integer(((CacheManager)cache).getPort(slave)).intValue());
		slaveConnection.writeData(dest, ms.serialize());
		
		//handle the response from slave
		IMessage dataResponse = null;
		try {
			dataResponse = slaveConnection.receiveMessage();
		} 
		catch (IOException e) {
			e.printStackTrace();
		}
		DataObject retval = dataResponse.getValue();
		terminateSlaveConnection();
		return retval;
	}
	
	public void terminateSlaveConnection()
	{
		IMessage msg = new Message();
		msg.setMessageID(IMessage.CLIENT_TERMINATE);
		msg.setDest(null);
		msg.setSource(this);
		msg.setKey(new DataObject("".getBytes()));
		msg.setValue(new DataObject("".getBytes()));
		slaveConnection.writeData(master, msg.serialize());		
		//slaveConnection.disconnectFrom(null);
	}

	@Override
	public int getHash(DataObject key) {
		int retVal = 0;
		// TODO Auto-generated method stub
		md.reset();
		byte[] digest = md.digest(key.getVal());
		DataInputStream is = new DataInputStream(new ByteArrayInputStream(digest));
		try {
			int a = is.readUnsignedByte();
			int b = is.readUnsignedByte();
			int c = is.readUnsignedByte();
			int d = is.readUnsignedByte();
			retVal = (((a & 0x3f) << 24) | ((b & 0xff) << 16) |
					  ((c & 0xff) << 8) | (d & 0xff));

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return retVal;
	}

	@Override
	public void putData(DataObject key, DataObject value) {
		int hashedKey = getHash(key);		
		String slave = cache.get(key);
		if(slave == null){
			//need to contact the master and get the corresponding slave
			IMessage message = new Message();
			message.setSource(this);
			message.setDest(master);
			message.setMessageID(IMessage.CLIENT_INQUIRE);
			message.setKey(key);
			message.setValue(new DataObject(new String("").getBytes()));
			masterConnection.writeData(master, message.serialize());
			IMessage response = null;
			try {
				response = masterConnection.receiveMessage();
				populateKeySpace(response.getValue().getVal());
			}
			catch (IOException e) {
				e.printStackTrace();
			}
			KeySpaceObject sl = findSlave(hashedKey); 
			slave = sl.getSlaveip();
			cache.put(key, slave); //the response contains the hashed key
			((CacheManager)cache).putPort(sl.getSlaveip(), sl.getPort());
		}
		//we should have got the slave by now. now contact the slave directly
		IMessage ms = new Message();
		ms.setSource(this);
		ms.setMessageID(IMessage.PUTDATA);
		ms.setKey(key);
		ms.setValue(value);
		IMachine dest = new Machine();
		dest.setIp(slave);
		ms.setDest(dest);
		slaveConnection.connectTo(dest,new Integer(((CacheManager)cache).getPort(slave)).intValue());
		slaveConnection.writeData(dest, ms.serialize());
		terminateSlaveConnection();		
	}
	
	public void closeAllConnections()
	{
		IMessage msg = new Message();
		msg.setMessageID(IMessage.CLIENT_TERMINATE);
		msg.setDest(master);
		msg.setSource(this);
		msg.setKey(new DataObject("".getBytes()));
		msg.setValue(new DataObject("".getBytes()));
		masterConnection.writeData(master, msg.serialize());
		masterConnection.disconnectFrom(master);
		slaveConnection.disconnectFrom(null);
	}
	
	public static void main(String[] args) throws IOException//first argument should be the ip of the client 
	{
		Client c = new Client();
		c.setIp(args[0]);
		c.setId(args[0]);
		int i = 0;
		while(i < 5){
			c.putData(new DataObject(new String("abcd").getBytes()),new DataObject(new String("world").getBytes()));
			String val = new String((c.getData(new DataObject(new String("abcd").getBytes()))).getVal());
			System.out.println("value got for key hello is " + val);
			System.in.read();
			i++;
		}
		c.closeAllConnections();
	}

}
